#include "stream/impl/rocksdb/Rocksdb.h"

#include <cassert>
#include <filesystem>
#include <iostream> // for cerr

#include "rocksdb/db.h"
#include "rocksdb/options.h"
#include "stream/impl/rocksdb/Table.h"

namespace stream {
namespace impl {
namespace rocksdb {

Rocksdb::Rocksdb(const std::string & name)
: _name(name)
, _dbOptions(new ::rocksdb::DBOptions())
, _cfOptions(new ::rocksdb::ColumnFamilyOptions())
, _openOptions(new ::rocksdb::Options())
, _readOptions(new ::rocksdb::ReadOptions())
, _writeOptions(new ::rocksdb::WriteOptions())
{
}
Rocksdb::~Rocksdb() {
	this->close();

	delete _writeOptions;
	delete _readOptions;

	delete _openOptions;

	delete _cfOptions;
	delete _dbOptions;
}

bool Rocksdb::exist(const std::string & dbname) {
	return std::filesystem::exists(dbname);
}
bool Rocksdb::touch(const std::string & dbname){
	::rocksdb::DB* rdb;
	::rocksdb::Options options;
	options.create_if_missing = true;
	auto s = ::rocksdb::DB::Open(options, dbname, &rdb);
	if(s.ok()){
		rdb->Close(); 
		delete rdb;
		return true;
	}
	std::cerr << "touch db failed:" << s.ToString() << std::endl;
	return false;
}

bool Rocksdb::isOpened() const{
	return _isOpened;
}


bool Rocksdb::createIfMissing() {
	if(!_isOpened){
		if(!Rocksdb::exist(_name)){
			if(false == Rocksdb::touch(_name)){
				return false;
			}
		}
		if(_openExist()){ //open all Column Families ,save <name, handle>=>tables
			_isOpened = true;
			return true;
		}
		return false;
	}
	return true;
}
//open exist  
//open all column families
bool Rocksdb::openExist() {
	if(!_isOpened){
		assert(Rocksdb::exist(_name)); 
		if(_openExist()){ //open all Column Families ,save <name, handle>=>tables
			_isOpened = true;
			return true;
		}
		return false;
	}
	return true;
}
void Rocksdb::close() {
	if(_isOpened){
		//TODO: destroy all column families ????? is need or not ???
//			for(auto && [name, handle] : _tables){
//				this->_closeColumnFamilyHandle(handle);	
//			}

		_db->Close();
		_db.reset();
		_isOpened = false;
	}
}
const std::map<std::string, ::rocksdb::ColumnFamilyHandle *> & Rocksdb::tables() const{
	return _tables;
}

std::pair<bool, ::rocksdb::ColumnFamilyHandle*> Rocksdb::isTableExist(const std::string & name) const{
	auto iter = _tables.find(name);
	if(iter != _tables.end()){ //find it
		return std::make_pair(true, iter->second);
	}
	return std::make_pair(false, (::rocksdb::ColumnFamilyHandle *)nullptr);

}
std::shared_ptr<Table> Rocksdb::getTable(const std::string & name){
	auto pexist = isTableExist(name);
	if(pexist.first){
		return std::make_shared<Table>(_db, name, pexist.second);
	}
	return std::shared_ptr<Table>();
}
std::shared_ptr<Table> Rocksdb::createTable(const std::string & name){
	auto pexist  = isTableExist(name);
	if(!pexist.first){ // not exist 
		::rocksdb::ColumnFamilyHandle * handle  = nullptr;
		auto s = _db->CreateColumnFamily(*_cfOptions, name, &handle);
		if(!s.ok()){
			std::cerr << "create column family failed:" << s.ToString() << std::endl;
			return std::shared_ptr<Table>();
		}

		_tables.insert(std::make_pair(name, handle));

		return std::make_shared<Table>(_db, name, handle);
	}
	return std::make_shared<Table>(_db, name, pexist.second);
}
bool Rocksdb::dropTable(const std::string & name){
	auto pexist = isTableExist(name);
	if(pexist.first){
		_tables.erase(name);

		auto s = _db->DropColumnFamily(pexist.second);
		if(!s.ok()){
			std::cerr << "ERROR: drop column family" << s.ToString() << std::endl;
			return false;	
		}
		return _closeColumnFamilyHandle(pexist.second);
	}
	return true;
}

bool Rocksdb::_closeColumnFamilyHandle(::rocksdb::ColumnFamilyHandle * handle){
	assert(handle != nullptr);

	auto s =_db->DestroyColumnFamilyHandle(handle);
	if(!s.ok()){
		std::cerr << "destroy column family" << s.ToString() << std::endl;
		return false;
	}
	return true;
}

bool Rocksdb::_openExist(){
	//1. get all columns  from exist db & save the tables handle
	std::vector<std::string> cfNames;
	auto s = ::rocksdb::DB::ListColumnFamilies(*_dbOptions, _name, &cfNames);
	if(!s.ok()) {  
		std::cerr << s.ToString() << std::endl;
		return false;
	}
	//2. create ColumnFamilyDescriptors from cfNames
	std::vector<::rocksdb::ColumnFamilyDescriptor> cfDescriptors;
	for(auto && name : cfNames){
		cfDescriptors.push_back(
			::rocksdb::ColumnFamilyDescriptor(name, *_cfOptions));
	}

	//3. open the db with cfDescriptors , return cfHandles
	::rocksdb::DB* rdb = nullptr;
	std::vector<::rocksdb::ColumnFamilyHandle*> cfHandles;
	s = ::rocksdb::DB::Open(*_openOptions, _name, cfDescriptors, &cfHandles, &rdb);
	if(!s.ok()){
		std::cerr << "open failed:" << s.ToString() << std::endl;
		return false;
	}
	_db.reset(rdb, [this](::rocksdb::DB * rdb) {
			std::cout << this->_name << " is deleted" << std::endl;
			delete rdb; 
	});
	//3. save the handles of column family
	for(auto && handle : cfHandles){
		_tables.insert(std::make_pair(handle->GetName(), handle));
	}
	return true;
}
::rocksdb::Status Rocksdb::put(const std::string & key, const std::string & value){
	return _db->Put(*_writeOptions, key, value);
}
::rocksdb::Status Rocksdb::get(const std::string & key, std::string & value){
	return _db->Get(*_readOptions, key, &value);
}
::rocksdb::Status Rocksdb::del(const std::string & key){
	return _db->Delete(*_writeOptions, key);
}
bool Rocksdb::size(size_t & sz) const{
	std::unique_ptr<::rocksdb::Iterator> iter(_db->NewIterator(*_readOptions));
	if(iter == nullptr){
		std::cerr << "ERROR: new iterator null" << std::endl;
		return false;	
	}
	for(sz = 0, iter->SeekToFirst(); iter->Valid(); iter->Next()){
		++sz;
	}
	return true;
}

}//end namespace rocksdb
}//end namespace impl
}//end namespace stream
